package com.gugee.ins.data.collector.bloggerfollower.service.impl;

import com.gugee.common.thread.ScheduleTemplate;
import com.gugee.ins.common.dal.service.blogger.InsBloggerDao;
import com.gugee.ins.common.dal.service.blogger.InsBloggerMainDao;
import com.gugee.ins.common.dal.service.blogger.InsRelBloggerFollowerDao;
import com.gugee.ins.common.dal.spider.InsBloggerSpiderClient;
import com.gugee.ins.common.model.mysql.blogger.InsBloggerMain;
import com.gugee.ins.common.model.spider.blogger.EdgeBlogger;
import com.gugee.ins.common.model.spider.blogger.InsBloggerFollowedResponse;
import com.gugee.ins.common.model.spider.blogger.PageInfo;
import com.gugee.ins.data.collector.bloggerfollower.config.InsBloggerFollowerConfig;
import com.gugee.ins.data.collector.bloggerfollower.service.InsBloggerFollowerService;
import com.gugee.ins.data.common.msg.InsBloggerMessage;
import com.xiguaji.boot.toolkit.ThreadPoolUtils;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.StringUtils;

import javax.annotation.PostConstruct;
import java.util.*;
import java.util.concurrent.ForkJoinPool;

/**
 * insBlogger服务
 * @author Huangxin
 */
@Slf4j
@Service
public class InsBloggerFollowerServiceImpl implements InsBloggerFollowerService {

    /**
     * 配置
     */
    @Autowired
    InsBloggerFollowerConfig insBloggerFollowerConfig;

    /**
     * 线程调度
     */
    @Autowired
    ScheduleTemplate scheduleTemplate;

    /**
     * 爬虫接口
     */
    @Autowired
    InsBloggerSpiderClient insBloggerSpiderClient;

    /**
     * InsBloggerDao
     */
    @Autowired
    InsBloggerDao insBloggerDao;

    /**
     * InsBloggerMainDao
     */
    @Autowired
    InsBloggerMainDao insBloggerMainDao;

    /**
     * InsRelBloggerFollowerDao
     */
    @Autowired
    InsRelBloggerFollowerDao insRelBloggerFollowerDao;

    /**
     * kafkaTemplate
     */
    @Autowired
    KafkaTemplate<Long,InsBloggerMessage> kafkaTemplateInsBlogger;

    /**
     * 线程池
     */
    ForkJoinPool forkJoinPool;

    /**
     * 线程名
     */
    private String workName = "insBloggerFollower-worker";

    /**
     * 初始化线程池
     */
    @PostConstruct
    public void init() {
        forkJoinPool = ThreadPoolUtils.forkJoinPool(insBloggerFollowerConfig.getInsBloggerFollowerCorePollSize(), workName);
        scheduleTemplate.attach(workName, forkJoinPool);
    }

    /**
     * 处理 播主的粉丝列表 采集 任务
     * @param insBloggerMessages
     */
    @Override
    public void handleInsBloggerFollower(List<InsBloggerMessage> insBloggerMessages) {
        try {
            forkJoinPool.submit(() -> insBloggerMessages.parallelStream().forEach(msg -> {
                try {
                    //采集播主关注人数据
                    runCollectFollower(msg);
                } catch (Exception ex) {
                    ex.printStackTrace();
                    log.error("ins blogger follower collect error. uid:{}, error :{}" ,
                            msg.getUid(),ex.getMessage());
                }
            })).get();
        } catch (Exception ex) {
            log.error("ins blogger follower collect  thread error.  {}" ,  ex.getMessage());
        }
    }

    /**
     * 采集 播主的粉丝列表
     * @param insBloggerMessage
     */
    public void runCollectFollower(InsBloggerMessage insBloggerMessage){
        long start = System.currentTimeMillis();
        long uid = insBloggerMessage.getUid();

        InsBloggerMain insBloggerMain = insBloggerMainDao.findByUid(uid);
        // 最近采集时间限制
        long lastCollectTime = start / 1000 - insBloggerFollowerConfig.getInsBloggerFollowerIntervalDay() * 24 * 3600;
        if(insBloggerMain == null || insBloggerMain.getIsPrivate()
                || insBloggerMain.getLastFollowerCollectTime() > lastCollectTime){
            return;
        }

        // 根据uid爬取信息
        PageInfo pageInfo;
        Set<Long> insRelBloggerFollowerUid = new HashSet<>(100);

        pageInfo = spiderFollower(uid,insBloggerFollowerConfig.getInsSpiderBloggerFollowerEachCount(),"",insRelBloggerFollowerUid);
        while(pageInfo != null && pageInfo.getHasNextPage()
                && !StringUtils.isEmpty(pageInfo.getEndCursor())
                && insRelBloggerFollowerUid.size() < insBloggerFollowerConfig.getInsBloggerFollowerCount()){
            pageInfo = spiderFollower(uid,insBloggerFollowerConfig.getInsSpiderBloggerFollowerEachCount(),pageInfo.getEndCursor(),insRelBloggerFollowerUid);
        }

        if(!insRelBloggerFollowerUid.isEmpty()){
            // 更新博主与粉丝的关系
            save(uid,insRelBloggerFollowerUid);
            // 更新标记位
            insBloggerMainDao.updateFollowerCollectTime(uid);
        }

        if ((System.currentTimeMillis() - start) / 1000 > insBloggerFollowerConfig.getInsBloggerFollowerDebugTimeLimit()) {
            log.info("ins blogger follower collect  uid:{} use time:{}s", insBloggerMessage.getUid(),(System.currentTimeMillis() - start) / 1000);
        }
    }

    /**
     * 保存
     * @param uid
     * @param insRelBloggerFollowerUid
     */
    @Transactional(rollbackFor = Exception.class)
    public void save(long uid,Set<Long> insRelBloggerFollowerUid){
        try{
            insRelBloggerFollowerDao.deleteByUid(uid);
            insRelBloggerFollowerDao.batchInsertByUid(uid,new ArrayList<>(insRelBloggerFollowerUid));
        }catch (Exception e){
            log.error("insert ins blogger follower collect error.msg : {}" ,e.getMessage());
        }
    }

    /**
     * 循环拉粉丝
     * @param uid
     * @param count
     * @param endCursor
     * @return
     */
    private PageInfo spiderFollower(long uid,int count,String endCursor,Set<Long> insRelBloggerFollowingUid){
        //根据uid爬取信息
        List<EdgeBlogger> edges;
        InsBloggerFollowedResponse insBloggerFollowedResponse;
        PageInfo pageInfo;
        try {
            insBloggerFollowedResponse = insBloggerSpiderClient.getFollowedBloggerByUid(uid,count,endCursor);
            edges = insBloggerFollowedResponse.getData().getUser().getEdgeFollow().getEdges();
            if(edges != null && !edges.isEmpty()){
                edges.forEach(edge -> {
                    if(edge != null && edge.getNode() != null
                            && !edge.getNode().getIsPrivate()
                            && edge.getNode().getUid() != null
                            && edge.getNode().getUid() > 0
                            && !insRelBloggerFollowingUid.contains(edge.getNode().getUid())){
                        insRelBloggerFollowingUid.add(edge.getNode().getUid());
                        sendPriorityCollectBloggerMsg(edge.getNode().getUid(),edge.getNode().getUsername(),edge.getNode().getPicUrl());
                    }
                });
            }
            pageInfo = insBloggerFollowedResponse.getData().getUser().getEdgeFollow().getPageInfo();
        }catch (Exception e){
            pageInfo = null;
            log.error("spider ins blogger follower error, msg :{}",e.getMessage());
        }
        return pageInfo;
    }

    /**
     * 发送博主优先收录的消息
     * @param uid
     * @param userName
     * @param picUrl
     */
    public void sendPriorityCollectBloggerMsg(long uid,String userName,String picUrl){
        if(uid > 0 && insBloggerDao.checkCollecting(uid,true)){
            //这里需要添加验证
            InsBloggerMessage insBloggerMessage = new InsBloggerMessage();
            insBloggerMessage.setUid(uid);
            insBloggerMessage.setUserName(userName);
            insBloggerMessage.setPicUrl(StringUtils.isEmpty(picUrl) ? "" : picUrl);
            insBloggerMessage.setRetryLimitTimes(insBloggerFollowerConfig.getInsBloggerPriorityRetryTimes());
            insBloggerMessage.setRetryTimes(insBloggerFollowerConfig.getInsBloggerPriorityRetryTimes());

            kafkaTemplateInsBlogger.send(insBloggerFollowerConfig.getInsBloggerPriorityCollectOnlineTopic(),insBloggerMessage);

        }
    }
}
